package com.tc.vms;

import com.tc.vms.producer.PubQoS;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.*;

import static com.tc.vms.message.MessageConst.*;
import static com.tc.vms.utils.Utils.MAX_DELAY_LEVEL;
import static com.tc.vms.utils.Utils.MIN_DELAY_LEVEL;

public class MessageWrap implements Serializable {
    private String messageId;
    private byte[] payloadData;
    private final List<String> routingKeys = new ArrayList<String>();
    private final HashMap<String, Object> properties = new HashMap<String, Object>();
    private final PubQoS pubQoS = new PubQoS();

    public MessageWrap() {
        properties.put(PROPERTY_CONTENT_ENCODING, VMS_MESSAGE_ENCODING);
        properties.put(PROPERTY_CONTENT_TYPE, VMS_MESSAGE_CONTENT_TYPE);
        properties.put(PROPERTY_BORN_HOST, HOSTNAME);
        properties.put(PROPERTY_BORN_TIMESTAMP, String.valueOf(System.currentTimeMillis()));
    }

    public MessageWrap(byte[] data) {
        this();
        this.payloadData = data;
    }

    public MessageWrap durable(boolean durable) {
        pubQoS.durable(durable);
        return this;
    }

    public boolean durable() {
        return pubQoS.durable();
    }

    public MessageWrap mandatory(boolean mandatory) {
        pubQoS.mandatory(mandatory);
        return this;
    }

    public boolean mandatory() {
        return pubQoS.mandatory();
    }

    long getLength() {
        return payloadData == null ? 0 : payloadData.length;
    }

    public String contentEncoding() {
        return (String) properties.get(PROPERTY_CONTENT_ENCODING);
    }

    public String contentType() {
        return (String) properties.get(PROPERTY_CONTENT_TYPE);
    }

    public MessageWrap contentEncoding(String contentEncoding) {
        properties.put(PROPERTY_CONTENT_ENCODING, contentEncoding);
        return this;
    }

    public MessageWrap contentType(String contentType) {
        properties.put(PROPERTY_CONTENT_TYPE, contentType);
        return this;
    }

    public long getTtl() {
        long ttl = -1;
        if (!properties.containsKey(PROPERTY_TTL)) return ttl;
        try {
            ttl = Long.parseLong((String) properties.get(PROPERTY_TTL));
        } catch (Exception e) {
        }
        return ttl;
    }

    public MessageWrap setTtl(long ttl) {
        properties.put(PROPERTY_TTL, String.valueOf(ttl));
        return this;
    }

    public int getReconsumeTimes() {
        int reconsumeTimes = 0;
        if (!properties.containsKey(PROPERTY_RECONSUME_TIME)) return reconsumeTimes;
        try {
            reconsumeTimes = Integer.parseInt((String) properties.get(PROPERTY_RECONSUME_TIME));
        } catch (Exception e) {
        }
        return reconsumeTimes;
    }

    protected MessageWrap setReconsumeTimes(int times) {
        properties.put(PROPERTY_RECONSUME_TIME, String.valueOf(times));
        return this;
    }

    public MessageWrap setBornHost(String host) {
        properties.put(PROPERTY_BORN_HOST, host);
        return this;
    }

    public String getBornHost() {
        try {
            return (String) properties.get(PROPERTY_BORN_HOST);
        } catch (Exception e) {
            return "";
        }
    }

    public MessageWrap setStoreTimestamp(long storeTimestamp) {
        properties.put(PROPERTY_STROE_TIMESTAMP, String.valueOf(storeTimestamp));
        return this;
    }

    public long getStoreTimestamp() {
        long stornTime = -1L;
        try {
            stornTime = Long.parseLong((String) properties.get(PROPERTY_STROE_TIMESTAMP));
        } catch (Exception e) {
        }
        return stornTime;
    }

    public MessageWrap setUserId(String userId) {
        properties.put(PROPERTY_USER_ID, userId);
        return this;
    }

    public String getUserId() {
        if (properties.containsKey(PROPERTY_USER_ID)) {
            return String.valueOf(properties.get(PROPERTY_USER_ID));
        }
        return "";
    }

    public MessageWrap setAppId(String appId) {
        properties.put(PROPERTY_APP_ID, appId);
        return this;
    }

    public String getAppId() {
        if (properties.containsKey(PROPERTY_APP_ID)) {
            return String.valueOf(properties.get(PROPERTY_APP_ID));
        }
        return "";
    }

    public int getPriority() {
        int priority = 0;
        if (!properties.containsKey(PROPERTY_PRIORITY)) return priority;
        try {
            priority = Integer.parseInt((String) properties.get(PROPERTY_PRIORITY));
        } catch (Exception e) {
        }
        return priority;
    }

    public MessageWrap setPriority(int priority) {
        properties.put(PROPERTY_PRIORITY, String.valueOf(priority));
        return this;
    }

    public MessageWrap setBornTime(long timestamp) {
        properties.put(PROPERTY_BORN_TIMESTAMP, String.valueOf(timestamp));
        return this;
    }

    public MessageWrap setDelayLevel(int delayLevel) {
        int tmpDelayLvl = delayLevel > MAX_DELAY_LEVEL ? MAX_DELAY_LEVEL : delayLevel;
        tmpDelayLvl = delayLevel < MIN_DELAY_LEVEL ? MIN_DELAY_LEVEL : delayLevel;
        properties.put(PROPERTY_DELAY_TIME_LEVEL, String.valueOf(tmpDelayLvl));
        return this;
    }

    public int getDelayLevel() {
        int delayLevel = 0;
        if (!properties.containsKey(PROPERTY_DELAY_TIME_LEVEL)) return delayLevel;
        try {
            delayLevel = Integer.parseInt((String) properties.get(PROPERTY_DELAY_TIME_LEVEL));
            delayLevel = delayLevel >= MAX_DELAY_LEVEL ? MAX_DELAY_LEVEL : delayLevel;
            delayLevel = delayLevel < MIN_DELAY_LEVEL ? MIN_DELAY_LEVEL : delayLevel;

        } catch (Exception e) {
        }
        return delayLevel;
    }

    public long getBornTime() {
        long bornTime = 0;
        if (!properties.containsKey(PROPERTY_BORN_TIMESTAMP)) return bornTime;
        try {
            bornTime = Long.parseLong((String) properties.get(PROPERTY_BORN_TIMESTAMP));
        } catch (Exception e) {
        }
        return bornTime;
    }

    /**
     * Sets the message id,which could be dedup message
     *
     * @param messageId
     * @return
     */
    public MessageWrap setMessageId(String messageId) {
        this.messageId = messageId;
        return this;
    }

    public MessageWrap setAttribute(String key, Object value) {
        properties.put(key, value);
        return this;
    }

    public Object getAttribute(String key) {
        return properties.get(key);
    }

    /**
     * Adds routing keys,which will be combined with dot in rabbitmq.
     *
     * @param keys The routing key pattern.
     * @return
     */
    public MessageWrap addRoutingKey(String... keys) {
        Collections.addAll(routingKeys, keys);
        return this;
    }

    /**
     * Gets the message id which's set by user.
     *
     * @return
     */
    public String getMessageId() {
        return messageId == null ? "" : messageId;
    }

    public Set<Map.Entry<String, Object>> getProperties() {
        return properties.entrySet();
    }

    public List<String> getRoutingKeys() {
        return routingKeys;
    }

    public byte[] getPayload() {
        return payloadData;
    }

    public MessageWrap setPayloadData(byte[] payloadData) {
        this.payloadData = payloadData;
        return this;
    }

    public static MessageWrap from(String message) {
        try {
            return new MessageWrap(message.getBytes(VMS_MESSAGE_ENCODING));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return null;
    }

    public static MessageWrap from(String message, String routingKey) {
        try {
            MessageWrap vmsMsg = new MessageWrap(message.getBytes(VMS_MESSAGE_ENCODING));
            vmsMsg.addRoutingKey(routingKey);
            return vmsMsg;
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return null;
    }

}
